import sys
import os
# 获取当前文件所在目录的父目录（即项目根目录）
project_root = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
if project_root not in sys.path:
    sys.path.append(project_root)
import json
import time
from kafka import KafkaProducer,KafkaConsumer
from common.start import bilibili_logger,douyin_logger,kuaishou_logger,config
class KafkaClient:
    def __init__(self, broker_urls: list, topic: str):
        """
        初始化Kafka客户端
        """
        self.broker_urls = broker_urls
        self.topic = topic
        self.max_retries = 3
        self.producer = None

    def connect(self):
        """
        连接Kafka集群
        """
        self.producer = KafkaProducer(
            bootstrap_servers=self.broker_urls,
            acks='all',  # 确保消息被所有副本接收
            retries=3,   # 发送失败时的重试次数
            value_serializer=lambda v: v.encode('utf-8'),  # 自动将字符串编码为bytes
            request_timeout_ms=30000,  # 超时时间设为30秒
            retry_backoff_ms=1000,     # 重试间隔1秒
            max_block_ms=60000,        # 阻塞最大时长60秒

        )
        return self  # 支持链式调用

    def send_message(self, message: str):
        """
        同步发送消息到Kafka
        :param message: 要发送的消息内容
        """
        retries = 0 
        while retries < self.max_retries:
            try:
                # 同步发送（通过get()方法等待结果）
                future = self.producer.send(self.topic, value=message)
                record_metadata = future.get(timeout=3)  # 等待最多10秒
                return
            except Exception as e:
                print(f"Failed to send message to Kafka: {e}")
                retries += 1


    def close(self):
        """
        关闭生产者连接
        """
        if self.producer:
            self.producer.flush()  # 确保所有消息都已发送
            self.producer.close()
            self.producer = None

    def __enter__(self):
        """支持with语句"""
        self.connect()
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        """支持with语句"""
        self.close()

BROKER_URLS = config["kafka"]["broker_urls"]
print(BROKER_URLS)
kafka_posts_client = KafkaClient(broker_urls=BROKER_URLS, topic="dap_koc_posts")
kafka_creators_client = KafkaClient(broker_urls=BROKER_URLS, topic="dap_koc_creators")
def peek_first_kafka_message(bootstrap_servers: list, topic: str, auto_offset_reset: str = 'earliest'):
    consumer = None
    try:
        # 创建临时消费者，禁用自动提交和组管理
        consumer = KafkaConsumer(
            topic,
            bootstrap_servers=bootstrap_servers,
            auto_offset_reset=auto_offset_reset,
            enable_auto_commit=False,  # 禁用自动提交
            group_id=None,  # 不使用消费者组(独立消费者)
            value_deserializer=lambda x: x.decode('utf-8'),  # 自动解码为字符串
            request_timeout_ms=30000,  # 超时时间设为30秒
            retry_backoff_ms=1000,     # 重试间隔1秒
        )
        
        print(f"尝试从topic '{topic}'读取第一条消息...")
        
        # 获取第一条消息
        message = next(consumer)
        print("\n=== 第一条消息 ===")
        print(f"Topic: {message.topic}")
        print(f"Partition: {message.partition}")
        print(f"Offset: {message.offset}")
        print(f"消息内容: {message.value}")
        print("=================\n")
        
        return message.value
        
    except StopIteration:
        print(f"Topic '{topic}'中没有消息")
        return None
    except Exception as e:
        print(f"未知错误: {e}")
        return None
    finally:
        if consumer is not None:
            consumer.close()
            print("消费者连接已关闭")

def send_bilibili_user_detail_message_to_kafka(up_detail):
    kafka_creators_client.connect()
    gender = ""
    if up_detail["sex"] == "女":
        gender = "female"
    elif up_detail["sex"] == "男":
        gender = "male"
    else:
        gender = "unknown"

    data = {
        "media" : "bilibili",
        "creator_id" : up_detail["uid"],
        "creator_name": up_detail["nick_name"],
        "home_page_url":  up_detail["link"],
        "avatar_url": up_detail["avatar"],
        "ip": "",
        "ip_geo": "",
        "follower_count": str(up_detail["fans"]),
        "mcn": "",
        "email":  "",
        "gender": gender,
        "register_time": str(up_detail["register_date"]),
        "tags": str(up_detail["tags"]),
        "seasons_series": str(up_detail["seasons_series"]),
        "origin_id": "",
        "create_time": int(time.time()),
    }
    print(json.dumps(data))
    kafka_creators_client.send_message(json.dumps(data))
    kafka_creators_client.close()

def send_bilibili_video_details_message_to_kafka(video_details, up_detail):
    kafka_posts_client.connect()
    gender = ""
    if up_detail["sex"] == "女":
        gender = "female"
    elif up_detail["sex"] == "男":
        gender = "male"
    else:
        gender = "unknown"

    for video_detail in video_details:
        data = {
            "media" : "bilibili",
            "creator_id": video_detail["uid"],
            "post_id": video_detail["bvid"],
            "post_title": video_detail["title"],
            "post_url": video_detail["video_link"],
            "view_count":  str(video_detail["view"]),
            "like_count": str(video_detail["like"]),
            "share_count": str(video_detail["share"]),
            "comment_count": str(video_detail["reply"]),
            "forward_count": "0",
            "dislike_count": "0",
            "collect_count": str(video_detail["favorite"]),
            "coin_count":  str(video_detail["coin"]),
            "danmu_count": str(video_detail["danmaku"]),
            "ip": "",
            "ip_geo": "",
            "followers": str(up_detail["fans"]),
            "mcn": "",
            "email": "",
            "gender": gender,
            "post_time":  str(video_detail["pubdate"]),
            "tags": str(video_detail["tags"]),
            "duration": str(video_detail["duration"]),
            "origin_id": "",
            "create_time": int(time.time()),
        } 
        kafka_posts_client.send_message(json.dumps(data))
    bilibili_logger.info(f"{len(video_details)} videos were sent to kafka")
    kafka_posts_client.close()


def send_kuaishou_user_detail_message_to_kafka(user_detail):
    kafka_creators_client.connect()
    gender = "unknown"
    if user_detail["gender"] == 'M':
        gender = "男"
    elif user_detail["gender"] == 'F':
        gender = "女"

    data = {
        "media" : "kuaishou",
        "creator_id" : user_detail["user_id"],
        "creator_name": user_detail["user_name"],
        "home_page_url":  user_detail["home_url"],
        "avatar_url": user_detail["avatar_url"],
        "ip": "",
        "ip_geo": "",
        "follower_count": user_detail["fans"],
        "mcn": "",
        "email":  "",
        "gender": gender,
        "register_time": user_detail["register_time"],
        "tags": user_detail["tags"],
        "seasons_series": "",
        "origin_id": "",
        "create_time": int(time.time()),
    }
    print(json.dumps(data))
    kafka_creators_client.send_message(json.dumps(data))
    kafka_creators_client.close()

def send_kuaishou_video_details_message_to_kafka(video_details, user_detail):
    kafka_posts_client.connect()
    gender = "unknown"
    if user_detail["gender"] == 'M':
        gender = "男"
    elif user_detail["gender"] == 'F':
        gender = "女"

    for video_detail in video_details:
        data = {
            "media" : "kuaishou",
            "creator_id": video_detail["kuaishou_user_id"],
            "post_id": video_detail["video_id"],
            "post_title": video_detail["title"],
            "post_url": video_detail["video_link"],
            "view_count":  video_detail["play_count"],
            "like_count": video_detail["like_count"],
            "share_count": "0",
            "comment_count": "0",
            "forward_count": "0",
            "dislike_count": "0",
            "collect_count": "0",
            "coin_count": "0",
            "danmu_count": "0",
            "ip": "",
            "ip_geo": "",
            "followers": user_detail["fans"],
            "mcn": "",
            "email": "",
            "gender": gender,
            "post_time":  video_detail["create_time"],
            "tags": video_detail["tags"],
            "duration": video_detail["duration"],
            "origin_id": "",
            "create_time": int(time.time()),
        }
        kafka_posts_client.send_message(json.dumps(data))
    kuaishou_logger.info(f"{len(video_details)} videos were sent to kafka")
    kafka_posts_client.close()

def send_douyin_user_detail_message_to_kafka(user_detail):
    kafka_creators_client.connect()
    gender = "unknown"

    data = {
        "media" : "douyin",
        "creator_id" : user_detail["sec_uid"],
        "creator_name": user_detail["nickname"],
        "home_page_url":  user_detail["home_page_url"],
        "avatar_url": user_detail["avatar_url"],
        "ip": "",
        "ip_geo": user_detail["ip_location"],
        "follower_count": str(user_detail["follower_count"]),
        "mcn": "",
        "email":  "",
        "gender": gender,
        "register_time": user_detail["register_time"],
        "tags": user_detail["tags"],
        "seasons_series": "",
        "origin_id": "",
        "create_time": int(time.time()),
    }
    print(json.dumps(data))
    kafka_creators_client.send_message(json.dumps(data))
    kafka_creators_client.close()

def send_kuaishou_video_details_message_to_kafka(video_details, user_detail):
    kafka_posts_client.connect()
    gender = "unknown"

    for video_detail in video_details:
        data = {
            "media" : "douyin",
            "creator_id": video_detail["sec_uid"],
            "post_id": video_detail["aweme_id"],
            "post_title": video_detail["title"],
            "post_url": video_detail["video_link"],
            "view_count": "0",
            "like_count": str(video_detail["like"]),
            "share_count": str(video_detail["share"]),
            "comment_count": str(video_detail["comment"]),
            "forward_count": "0",
            "dislike_count": "0",
            "collect_count": str(video_detail["collect"]),
            "coin_count": "0",
            "danmu_count": "0",
            "ip": "",
            "ip_geo": user_detail["ip_location"],
            "followers": str(user_detail["follower_count"]),
            "mcn": "",
            "email": "",
            "gender": gender,
            "post_time":  str(video_detail["create_time"]),
            "tags": str(video_detail["tags"]),
            "duration": str(video_detail["duration"]),
            "origin_id": str(user_detail["origin_id"]),
            "create_time": int(time.time()),
        }
        kafka_posts_client.send_message(json.dumps(data))
    douyin_logger.info(f"{len(video_details)} videos were sent to kafka")
    kafka_posts_client.close()

if __name__ == "__main__":
    # data = {
    #     "media" : "bilibili",
    #     "title" : "测试",
    #     "create_time": int(time.time())
    # }
    # print(json.dumps(data))
    servers = BROKER_URLS  # 替换为你的Kafka服务器地址
    topic = "dap_koc_creators"          # 替换为你的topic名称
    
    # 读取第一条消息
    first_message = peek_first_kafka_message(servers, topic, auto_offset_reset='earliest')
    
    if first_message is not None:
        print("成功读取到第一条消息:")
        print(json.loads(first_message))
    else:
        print("未能读取到消息")